package com.shiguiwu.springboot3.kafka.controller;

import cn.hutool.json.JSONUtil;
import com.shiguiwu.springboot3.util.R;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @description: kafka
 * @author: stone
 * @date: Created by 2023/11/13 16:31
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springboot3.kafka.controller
 */
@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/sendSimpleMessage")
    public R<String> sendSimpleMessage(@RequestParam(value = "topic", defaultValue = "test520") String topic, @RequestParam("msg") String msg) throws ExecutionException, InterruptedException {
        for (int i = 0; i < 1000; i++) {
            CompletableFuture<SendResult<String, Object>> f = kafkaTemplate.send(topic, msg + i);
            SendResult<String, Object> result = f.get();
            log.info("发送结果 ==> {}", result.getRecordMetadata().partition());
        }
        return R.ok("发送成功");
    }

    /**
     * 事务
     * @param topic
     * @param msg
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @GetMapping("/transaction")
    public R<String> transaction(@RequestParam(value = "topic", defaultValue = "test520") String topic, @RequestParam("msg") String msg) throws ExecutionException, InterruptedException {
        //声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            return operations.send(topic, msg + " test executeInTransaction");
            //throw new RuntimeException("fail");
        });

        return R.ok("失败以后，消息不投递了");
    }

    /**
     * 带回调的
     * @param topic
     * @param msg
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @GetMapping("/callbackOneMessage")
    public R<String> callbackOneMessage(@RequestParam(value = "topic", defaultValue = "test520") String topic, @RequestParam("msg") String msg) throws ExecutionException, InterruptedException {
        //失败的回调
        //成功的回调
        kafkaTemplate.send(topic, msg)
                .thenAccept(result -> {
                    RecordMetadata resultRecordMetadata = result.getRecordMetadata();
                    log.warn("发送成功:{}", JSONUtil.toJsonStr(result));
                })
                .exceptionally(e -> {
                    log.error("发生错误 ", e);
                    return null;
                });
        return R.ok("发送成功");
    }


    @GetMapping("/sendMulti")
    public R<String> sendMulti(@RequestParam(value = "topic", defaultValue = "shigwTopic") String topic, @RequestParam("msg") String msg) throws ExecutionException, InterruptedException {
        for (int i = 0; i < 100; i++) {
            CompletableFuture<SendResult<String, Object>> f = kafkaTemplate.send(topic, msg + i);
            SendResult<String, Object> result = f.get();
            log.info("发送结果 ==> {}", result.getRecordMetadata().partition());
        }
        return R.ok("发送成功");
    }
}
